package bsf.baseservice.businessmq.consumer;

import bsf.base.CallBack;
import bsf.baseservice.businessmq.db.*;
import bsf.baseservice.businessmq.redis.*;
import bsf.baseservice.businessmq.systemruntime.*;
import bsf.baseservice.businessmq.systemruntime.log.*;
import bsf.db.DbConn;
import bsf.serialization.json.JsonProvider;
import bsf.system.thread.ThreadPool;
import bsf.util.ConvertUtil;

import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/** 
 消费者心跳守护
 
*/
public class ConsumerHeartbeatProtect implements AutoCloseable
{
	private ConsumerContext Context;
	private Date lastupdatetimeofmqpath = new Date(0);
	private Object heartbeatrunlock = new Object();
	private RedisNetCommandListener redislistener = null;
	private boolean isClosed = false;
	public ConsumerHeartbeatProtect(ConsumerContext context)
	{
		try
		{
			isClosed= false;
			Context = context;
			DebugHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "ConsumerHeartbeatProtect", "消费者准备心跳注册");

			lastupdatetimeofmqpath = getLastUpdateTimeOfMqPath();
			DebugHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "ConsumerHeartbeatProtect", "消费者获取队列最后更新时间成功");

			ThreadPool.SystemThreadPool.submit(new Runnable() {//开启心跳检查
				@Override
				public void run() {
					heartBeatRun();
				}
			});
			DebugHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "ConsumerHeartbeatProtect", "消费者开启心跳循环成功");

			redislistener = new RedisNetCommandListener(ConfigHelper.RedisServer);
			redislistener.Name="消费者" + context.ConsumerProvider.getClient();
			redislistener.register(new CallBack.Func2<Object,String,String>() {
				@Override
				public Object invoke(String channel, String message) {
					redisListenerCommand(channel, message);
					return null;
				}
			}
		   ,  Context.ConsumerProvider.getMQPath(), SystemParamConfig.Redis_Channel);
			DebugHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "ConsumerHeartbeatProtect", "消费者开启Redis消息订阅成功");
			LogHelper.writeLine(Context.ConsumerInfo.MQPathModel.id, Context.ConsumerInfo.MQPathModel.mqpath, "ConsumerHeartbeatProtect", "消费者心跳注册成功");
		}
		catch (RuntimeException exp)
		{
			ErrorLogHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "ConsumerHeartbeatProtect", "消费者心跳守护初始化", exp);
			throw exp;
		}
	}

	//心跳循环
	private void heartBeatRun()
	{
		try
		{
			while (isClosed==false)
			{
				heartBeatTask();
				try{Thread.sleep(SystemParamConfig.Consumer_HeatBeat_Every_Time * 1000);}catch (Exception exp){}
				DebugHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "ConsumerHeartbeatProtect-HeartBeatRun", "心跳循环一轮完毕");
			}
		}
		catch (RuntimeException exp)
		{
			ErrorLogHelper.writeLine(Context.ConsumerInfo.MQPathModel.id, Context.ConsumerInfo.MQPathModel.mqpath, "HeartBeatRun", "消费者心跳循环方法", exp);
		}
	}
	//心跳任务
	private void heartBeatTask()
	{
		synchronized (heartbeatrunlock)
		{
			try
			{
				//更新节点心跳
				ConsumerBLL bll = new ConsumerBLL();
				try(DbConn c = DbConn.createConn(Context.ConsumerProvider.Config.ManageConnectString))
				{
					bll.consumerHeartbeat(c, Context.ConsumerInfo.ConsumerModel.tempid, Context.ConsumerInfo.ConsumerModel.consumerclientid);
				}

				DebugHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "ConsumerHeartbeatProtect-HeartBeatTask", "更新节点心跳完毕");
				//检查当前队列是否有更新，有更新则重启customer
				Date lastupdatetime = getLastUpdateTimeOfMqPath();
				if (lastupdatetimeofmqpath.compareTo(lastupdatetime) < 0||Context.IsNeedReload == true)
				{
					Context.IsNeedReload=false;
					DebugHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "ConsumerHeartbeatProtect-HeartBeatTask", "检测到队列更新准备重启");
					//Context.ConsumerProvider.ReRegister();
					Context.ConsumerProvider.close();Context.ConsumerProvider.registerReceiveMQListener((CallBack.Action1)Context.ActionInfo.Action,Context.ActionInfo.ReturnType);
					/*java.lang.reflect.Method method = ConsumerProvider.class.getMethod("RegisterReceiveMQListener");
					method = method.MakeGenericMethod(Context.getActionInfo().getReturnType());
					method.invoke(Context.ConsumerProvider, new Object[] { Context.ActionInfo.Action });*/
					lastupdatetimeofmqpath = lastupdatetime;
					redislistener.RedisServerIp = ConfigHelper.RedisServer;
					DebugHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "ConsumerHeartbeatProtect-HeartBeatTask", "检测到队列更新重启成功");
					LogHelper.writeLine(Context.ConsumerInfo.MQPathModel.id, Context.ConsumerInfo.MQPathModel.mqpath, "队列更新重启", "消费者检测到队列更新重启成功");
				}
			}
			catch (Exception exp)
			{
				ErrorLogHelper.writeLine(Context.ConsumerInfo.MQPathModel.id, Context.ConsumerInfo.MQPathModel.mqpath, "HeartBeatTask", "消费者心跳循环出错", exp);
			}
		}

	}

	private void redisListenerCommand(String channel, String msg)
	{
		try
		{
			if (isClosed == false)
			{
			DebugHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "ConsumerHeartbeatProtect-RedisListenerCommand", "检测到redis消息,msg:"+ ConvertUtil.objToStr(msg));
			BusinessMQNetCommand command = new JsonProvider().deserialize(msg,BusinessMQNetCommand.class);
			if ((command.CommandReceiver == EnumCommandReceiver.All || command.CommandReceiver == EnumCommandReceiver.Consumer) && command.MqPath.toLowerCase().equals(Context.ConsumerProvider.getMQPath().toLowerCase()))
			{
				if (command.CommandType == EnumCommandType.Register)
				{
					DebugHelper.writeLine(Context.getMQPathID(), Context.getMQPath(), "ConsumerHeartbeatProtect-RedisListenerCommand", "检测到redis消息:" + EnumCommandType.Register.name());
					Context.IsNeedReload=true;
					heartBeatTask();
				}
			}
			}
		}
		catch (Exception exp)
		{
			ErrorLogHelper.writeLine(Context.ConsumerInfo.MQPathModel.id, Context.ConsumerInfo.MQPathModel.mqpath, "RedisListenerCommand", "消费者Redis注册监听", exp);
		}
	}

	private Date getLastUpdateTimeOfMqPath()
	{
		ConsumerBLL bll = new ConsumerBLL();
		try(DbConn c = DbConn.createConn(Context.ConsumerProvider.Config.ManageConnectString))
		{
			Date t = bll.getLastUpdateTimeOfMqPath(c, Context.ConsumerProvider.getMQPath());
			if(t == null)
			{
				throw new BusinessMQException(String.format("检测到队列%1$s最后更新时间为null", Context.ConsumerProvider.getMQPath()));
			}
			return t;
		}
	}

	public final void close()
	{
		if (redislistener != null)
		{
			try{redislistener.close();}catch (Exception exp){}
		}
		isClosed = true;

	}
}